package com.yixing.tech.common.tcp.manage;

import com.yixing.tech.common.tcp.channel.ChannelContext;
import com.yixing.tech.common.tcp.channel.ChannelInit;
import com.yixing.tech.common.tcp.client.NettyTcpClient;
import com.yixing.tech.common.tcp.handler.MessageHandler;
import com.yixing.tech.common.tcp.pc.TcpConsumer;
import com.yixing.tech.common.tcp.pc.TcpMessage;
import com.yixing.tech.common.tcp.pc.TcpProducer;
import com.yixing.tech.common.tcp.pc.TcpReply;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.function.Function;

@Slf4j
public class EndpointManager {

    private static final ConcurrentHashMap<String, TcpClientEndpoint> tcpClientEndpointMap = new ConcurrentHashMap<>();

    /** 新建tcpClientEndpoint端点 */
    public TcpClientEndpoint createTcpClientEndpoint(String remoteHost, int remotePort, Boolean isOpen, Function<TcpMessage, TcpReply> consumeFunc) {
        ChannelContext channelContext = new ChannelContext();

        MessageHandler messageHandler = new MessageHandler();
        messageHandler.setChannelContext(channelContext);

        log.info("创建生产者消费者");
        BlockingDeque<TcpMessage> deque = new LinkedBlockingDeque<>(100);
        TcpProducer tcpProducer = new TcpProducer(deque);

//        Function<TcpMessage, Void> func = msg -> {
//            tcpProducer.produce(msg);
//            return null;
//        };

//        messageHandler.setProduceFunc(func);
        messageHandler.setProducer(tcpProducer);
        ChannelInit channelInit = new ChannelInit(messageHandler);

//        Function<TcpMessage, TcpReply> consumeFunc = msg -> {
//            log.info("consume message:{}", msg);
//            log.info("TODO分发消息>>>>>>>>>>>>>>");
//
//            return new TcpReply(msg.getIp(), msg.getPort(), "11 11 11 22 22 22 03");
//        };
        TcpConsumer consumer = new TcpConsumer(deque, consumeFunc);
        consumer.setChannelContext(channelContext);

        log.info("创建netty客户端");
        NettyTcpClient nettyTcpClient = new NettyTcpClient(channelInit);
        nettyTcpClient.setRemoteHostAndPort(remoteHost, remotePort);

        log.info("创建端点");
        TcpClientEndpoint endpoint = new TcpClientEndpoint(nettyTcpClient);
        Thread thread = new Thread(consumer);
        thread.setName("tcpConsumer-endpoint[" + endpoint.getEid() + "]");
        thread.start();

        endpoint.setProducer(tcpProducer);
        endpoint.setConsumer(consumer);

        log.info("存储到上下文");
        tcpClientEndpointMap.put(endpoint.getEid(), endpoint);
        if(isOpen) {
            log.info("开始连接服务端");
            endpoint.connect(remoteHost, remotePort);
        }
        return endpoint;
    }

    public void closeTcpClientEndpoint(String eid) {
        TcpClientEndpoint clientEndpoint = tcpClientEndpointMap.get(eid);
        if(clientEndpoint != null) {
            clientEndpoint.disconnect();
            tcpClientEndpointMap.remove(eid);
        }
    }

    public void sendMsgByTcpClientEndpoint(String eid, String message) {
        TcpClientEndpoint clientEndpoint = tcpClientEndpointMap.get(eid);
        if(clientEndpoint != null) {
            clientEndpoint.send(message);
        }
    }

    public List<Map<String, Object>> listEndpoints() {
        List<Map<String, Object>> list = new ArrayList<>();
        for(Map.Entry<String, TcpClientEndpoint> entry : tcpClientEndpointMap.entrySet()) {
            Map<String, Object> map = new HashMap<>();
            map.put("eid", entry.getKey());
            map.put("connected", entry.getValue().isConnected());
            list.add(map);
        }
        return list;
    }
}
